package org.snail.transport.impl;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.HashedWheelTimer;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import org.snail.transport.Acceptor;
import org.snail.transport.AcceptorDispatcher;
import org.snail.transport.handler.*;

import java.util.concurrent.ThreadFactory;

/**
 * Created by fish on 17/8/26.
 */
public class NettyAcceptor implements Acceptor {


    private static final InternalLogger logger = InternalLoggerFactory.getInstance(NettyAcceptor.class);

    protected final HashedWheelTimer timer = new HashedWheelTimer(new DefaultThreadFactory("snail.acceptor.timer"));
    private final AcceptorIdleStateTrigger acceptorIdleStateTrigger = new AcceptorIdleStateTrigger();
    private final AcceptorHandler handler = new AcceptorHandler();
    protected final SnailEncoder encoder = new SnailEncoder();
    private ServerBootstrap bootstrap;
    private EventLoopGroup boss;
    private EventLoopGroup worker;

    @Override
    public void start(int port) {
        int workerNum = Runtime.getRuntime().availableProcessors() << 1;
        ThreadFactory bossFactory = new DefaultThreadFactory("snail.acceptor.boss", Thread.MAX_PRIORITY);
        boss =  new NioEventLoopGroup(1, bossFactory);

        ThreadFactory workerFactory = new DefaultThreadFactory("snail.acceptor.worker", Thread.MAX_PRIORITY);
        worker = new NioEventLoopGroup(workerNum, workerFactory);

        bootstrap = new ServerBootstrap().group(boss, worker);
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
        bootstrap.option(ChannelOption.SO_REUSEADDR, true);
        bootstrap.childOption(ChannelOption.SO_REUSEADDR, true)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.ALLOW_HALF_CLOSURE, false);

        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(
                        new IdleStateHandler(timer, 60, 0, 0),    //空闲链路监测,60秒.
                        acceptorIdleStateTrigger,                 //空闲链路监测
                        new SnailDecoder(),                       //解码
                        encoder,                                  //编码
                        handler);                                 //处理
            }
        });


        ChannelFuture future = bootstrap.bind(port);

        try {
            future.sync();
        } catch (InterruptedException e) {
            logger.error(e);
            this.shutdown();
        }

    }

    @Override
    public void bindDispatcher(AcceptorDispatcher dispatcher) {
        this.handler.setDispatcher(dispatcher);
    }

    @Override
    public void shutdown() {
        if(boss != null )boss.shutdownGracefully();
        if(worker != null)worker.shutdownGracefully();
    }
}
